package com.sh.data.engine.domain.integration.api.http.service.impl;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sh.data.engine.common.enumDefinition.ApiTaskStatus;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.common.util.JsonPathUtil;
import com.sh.data.engine.common.util.ServletUtils;
import com.sh.data.engine.domain.base.model.PageResult;
import com.sh.data.engine.domain.base.model.UserContext;
import com.sh.data.engine.domain.base.model.datax.ApiHttpJobContentReaderParameter;
import com.sh.data.engine.domain.base.model.datax.DataX;
import com.sh.data.engine.domain.base.model.datax.DataxReaderAndWriter;
import com.sh.data.engine.domain.base.model.enums.XxlGroupEnum;
import com.sh.data.engine.domain.common.loader.CustomerUrlClassLoader;
import com.sh.data.engine.domain.common.service.DistributedService;
import com.sh.data.engine.domain.dataasset.service.TblPropService;
import com.sh.data.engine.domain.datadev.offline.model.enums.OfflineDevJobStatusEnum;
import com.sh.data.engine.domain.datadev.offline.model.enums.OfflineDevTaskStopEnum;
import com.sh.data.engine.domain.integration.api.http.model.domain.ApiTaskDomain;
import com.sh.data.engine.domain.integration.api.http.model.domain.TaskDomain;
import com.sh.data.engine.domain.integration.api.http.model.domain.TaskOverviewDomain;
import com.sh.data.engine.domain.integration.api.http.model.domain.TestApiHttpDomain;
import com.sh.data.engine.domain.integration.api.http.model.mapper.HttpApiAuthParamMapper;
import com.sh.data.engine.domain.integration.api.http.model.mapper.HttpApiRequestBodyParamMapper;
import com.sh.data.engine.domain.integration.api.http.model.mapper.TestApiHttpRequestMapper;
import com.sh.data.engine.domain.integration.api.http.model.param.*;
import com.sh.data.engine.domain.integration.api.http.service.ApiHttpService;
import com.sh.data.engine.domain.integration.api.record.model.domain.ApiRunningRecordDomain;
import com.sh.data.engine.domain.integration.api.record.service.ApiRunningRecordService;
import com.sh.data.engine.domain.integration.api.taskconfig.model.domain.*;
import com.sh.data.engine.domain.integration.api.taskconfig.model.param.ExpressionParam;
import com.sh.data.engine.domain.integration.api.taskconfig.model.param.PreviewExpressParam;
import com.sh.data.engine.domain.integration.api.taskconfig.model.param.TaskConfigParam;
import com.sh.data.engine.domain.integration.api.taskconfig.model.param.TaskRecordParam;
import com.sh.data.engine.domain.integration.api.taskconfig.service.TaskConfigService;
import com.sh.data.engine.domain.integration.api.webservice.service.ApiWsService;
import com.sh.data.engine.domain.integration.datasource.model.domain.DataSourceDetailDomain;
import com.sh.data.engine.domain.integration.datasource.model.domain.DataSourceDetailDomain.RdbmsConfig;
import com.sh.data.engine.domain.integration.datasource.model.domain.DataSourceDomain;
import com.sh.data.engine.domain.integration.datasource.service.DataSourceService;
import com.sh.data.engine.domain.integration.offline.model.enums.OfflineSyncExecuTypeEnum;
import com.sh.data.engine.domain.integration.offline.service.DataXService;
import com.sh.data.engine.domain.shims.db.model.FieldInfoDomain;
import com.sh.data.engine.domain.shims.db.model.TableInfoDomain;
import com.sh.data.engine.domain.util.ConvertUtil;
import com.sh.data.engine.domain.util.TLSUtil;
import com.sh.data.engine.domain.util.VarParserUtil;
import com.sh.data.engine.domain.workspace.manager.model.domain.UserDomain;
import com.sh.data.engine.domain.workspace.manager.service.UserService;
import com.sh.data.engine.job.admin.controller.JobInfoController;
import com.sh.data.engine.job.admin.core.model.XxlJobInfo;
import com.sh.data.engine.job.admin.core.route.ExecutorRouteStrategyEnum;
import com.sh.data.engine.job.admin.core.scheduler.MisfireStrategyEnum;
import com.sh.data.engine.job.admin.core.scheduler.ScheduleTypeEnum;
import com.sh.data.engine.job.admin.service.XxlJobService;
import com.sh.data.engine.job.core.biz.model.ReturnT;
import com.sh.data.engine.job.core.enums.ExecutorBlockStrategyEnum;
import com.sh.data.engine.job.core.glue.GlueTypeEnum;
import com.sh.data.engine.repository.dao.base.EngineBaseModel;
import com.sh.data.engine.repository.dao.integration.api.http.ApiHttpMapper;
import com.sh.data.engine.repository.dao.integration.api.http.entity.ApiHttpEntity;
import com.sh.data.engine.repository.dao.integration.api.http.entity.ApiHttpEntity.ApiHttpRequest;
import com.sh.data.engine.repository.dao.integration.api.http.entity.ApiHttpEntity.ApiHttpRequestBodyParam.FormParam;
import com.sh.data.engine.repository.dao.integration.api.http.entity.ApiResponseEntity;
import com.sh.data.engine.repository.dao.integration.api.http.entity.TaskConfigParamEntity;
import com.sh.data.engine.repository.dao.integration.api.record.entity.ApiRunningRecordEntity;
import com.sh.data.engine.repository.dao.integration.api.webservice.entity.ApiWsEntity;
import com.sh.data.engine.repository.dao.integration.offline.entity.OfflineSyncEntity;
import lombok.extern.slf4j.Slf4j;
import okhttp3.*;
import okio.BufferedSource;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.reflect.MethodUtils;
import org.apache.commons.text.StringSubstitutor;
import org.apache.http.entity.ContentType;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.UnsupportedEncodingException;
import java.net.URL;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * @author : changkong
 * @date : 2022/3/11
 */
@Slf4j
@Service
public class ApiHttpServiceImpl extends ServiceImpl<ApiHttpMapper, ApiHttpEntity>
    implements ApiHttpService {

    @Autowired
    private TaskConfigService taskConfigServiceImpl;

    @Autowired
    private JobInfoController jobInfoController;

    @Autowired
    private XxlJobService xxlJobService;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private ApiRunningRecordService apiRunningRecordService;

    @Autowired
    private DataXService dataXService;

    @Autowired
    private UserService userService;

    @Autowired
    private DistributedService distributedService;

    @Autowired
    private ApiWsService apiWsService;

    @Autowired
    private TblPropService tblPropService;

    @Autowired
    private DataSourceService dataSourceService;

    @Autowired
    private TestApiHttpRequestMapper testApiHttpRequestMapper;

    @Autowired
    private HttpApiAuthParamMapper httpApiAuthParamMapper;

    @Autowired
    private HttpApiRequestBodyParamMapper httpApiRequestBodyParamMapper;

    private static final String varPattern = "\\$\\{(.+?)\\}";

    private static final Pattern pattern = Pattern.compile(varPattern);

    //该id在不同的环境不一样
    @Value("${super.admin.id}")
    private String SUPER_ADMIN;

    /**
     * http接口测试
     *
     * @param testApiHttpParam
     */
    @Override
    public TestApiHttpDomain testApiHttp(TestApiHttpParam testApiHttpParam) {
        String url = testApiHttpParam.getUrl();
        String httpRequestMethod = testApiHttpParam.getHttpRequestMethod();
        ApiHttpAuthParam apiHttpAuthParam = testApiHttpParam.getApiHttpAuthParam();

        ApiHttpRequestBodyParam apiHttpRequestBodyParam = testApiHttpParam.getApiHttpRequestBodyParam();
        String contentType = apiHttpRequestBodyParam.getContentType();
        List<ApiHttpRequestBodyParam.FormParam> formParams = apiHttpRequestBodyParam.getFormParams();
        String rawJson = apiHttpRequestBodyParam.getBody();
        // 处理auth参数 如替换token
        // 处理请求参数 如替换变量
        Request req =
            buildRequest(
                url, httpRequestMethod, apiHttpAuthParam, null, contentType, formParams, rawJson);
        return execRequest(req);
    }

    /**
     * 上传jar包
     *
     * @param multipartFile
     * @return
     */
    @Override
    public JarUploadResultDomain uploadJar(MultipartFile multipartFile) {
        return taskConfigServiceImpl.uploadJar(multipartFile);
    }

    /**
     * 预览
     *
     * @param previewExpressParam
     * @return
     */
    @Override
    public List<List<Object>> preview(PreviewExpressParam previewExpressParam) {
        return taskConfigServiceImpl.preview(previewExpressParam);
    }

    /**
     * 获取数据源列表
     *
     * @param projectId
     * @return
     */
    @Override
    public List<DataSourceDomain> getDatabases(Long projectId) {
        return taskConfigServiceImpl.getDatabases(projectId);
    }

    /**
     * 在数据源中选择目标表
     *
     * @param dsId
     * @return
     */
    @Override
    public List<TableInfoDomain> getTargetTables(Long dsId) {
        return taskConfigServiceImpl.getTargetTables(dsId);
    }

    /**
     * 根据字段获取建表语句
     *
     * @param projectId
     * @param fields
     * @return
     */
    @Override
    public String getCreateTableDDL(
        Long projectId, Long dsId, List<String> fields, Integer isSupportRtf) {
        return taskConfigServiceImpl.getCreateTableDDL(projectId, dsId, fields, isSupportRtf);
    }

    /**
     * 执行建表语句
     *
     * @param projectId
     * @param sql
     */
    @Override
    public String saveTargetTable(Long projectId, String sql) {
        return taskConfigServiceImpl.saveTargetTable(projectId, sql);
    }

    @Override
    public String saveTargetTable(Long projectId, String sql, Long dataSecretId) {
        return taskConfigServiceImpl.saveTargetTable(projectId, sql, dataSecretId);
    }

    /**
     * 获取目标表字段信息
     *
     * @param dsId
     * @param tableName
     * @return
     */
    @Override
    public List<FieldInfoDomain> getTargetTableFields(Long dsId, String tableName) {
        return taskConfigServiceImpl.getTargetTableFields(dsId, tableName);
    }

    /**
     * datax任务脚本配置
     *
     * @param reader
     * @param task
     * @return
     */
    @Override
    public String buildJobJson(DataX.JobContentReader reader, TaskConfigParam task) {
        return taskConfigServiceImpl.buildJobJson(reader, task);
    }

    /**
     * xxljob移除任务, mongo中删除任务
     *
     * @param id
     */
//    @BizDataIndexAnnotation(
//        type = BizDataIndexTypeEnum.API_HTTP_COLLECT,
//        dataIdFieldName = "id",
//        dataIdParamIndex = 0,
//        operateType = BizDataIndexAnnotation.OperateType.DELETE)
    @Override
    public void delete(Long id) {
        log.info("taskId:{}", id);
        ApiHttpEntity task = this.getById(id);
        if (null == task) {
            throw new BusinessException("检查任务状态");
        }
        if (null != task.getXxlJobId()) {
            Integer xxlJobId = task.getXxlJobId();
            xxlJobService.remove(xxlJobId);
        }
        task.setRowState(0);
        this.updateById(task);
    }

    @Override
    public Long run(Long taskId) {

        return dataXService.executeApiHttp(taskId, OfflineSyncExecuTypeEnum.MANUAL.getStatus(), null);
    }

    @Override
    public Long executeApiHttp(Long taskId, Integer executeType, Map<String, String> param) {
        Long logId = dataXService.executeApiHttp(taskId, executeType, param);
        return logId;
    }

    /**
     * 操作任务
     *
     * @param taskId
     * @param type
     */
    @Override
    public void operate(Long taskId, String type) {
        log.info("taskId:{},type:{}", taskId, type);
        ApiHttpEntity task = this.getById(taskId);
        if (null == task || null == task.getXxlJobId()) {
            throw new BusinessException("检查任务状态");
        }
        Integer xxlJobId = task.getXxlJobId();
        switch (type) {
            case "online":
                task.setTaskStatus(ApiTaskStatus.released.getTaskStatus());
                task.setUpdateTime(new Date());
                task.setUpdaterId(UserContext.getUserId());
                task.setStartTime(new Date());
                this.update(
                    task, new LambdaUpdateWrapper<ApiHttpEntity>().eq(ApiHttpEntity::getId, taskId));
                jobInfoController.start(xxlJobId);
                break;
            case "offline":
                task.setTaskStatus(ApiTaskStatus.torelease.getTaskStatus());
                task.setUpdateTime(new Date());
                task.setUpdaterId(UserContext.getUserId());
                task.setStartTime(null);
                this.update(
                    task, new LambdaUpdateWrapper<ApiHttpEntity>().eq(ApiHttpEntity::getId, taskId));
                jobInfoController.stop(xxlJobId);
                break;
            default:
                throw new BusinessException("不支持的操作");
        }
    }

    @Override
    public PageResult<TaskRecordDomain> historyTask(TaskRecordParam taskRecordParam) {
        return taskConfigServiceImpl.historyTask(taskRecordParam);
    }

    /**
     * 页面上方任务状态统计信息
     *
     * @param projectId
     * @return
     */
    @Override
    public TaskOverviewDomain getOverview(Long projectId) {
        LambdaQueryWrapper<ApiHttpEntity> toReleaseWrapper = new LambdaQueryWrapper<ApiHttpEntity>().eq(
                        ApiHttpEntity::getProjectId, projectId)
                .eq(ApiHttpEntity::getTaskStatus, ApiTaskStatus.torelease.getTaskStatus());
        long toReleaseCount = this.count(toReleaseWrapper);

        LambdaQueryWrapper<ApiRunningRecordEntity> lambdaQueryWrapper = new LambdaQueryWrapper<ApiRunningRecordEntity>().eq(
                ApiRunningRecordEntity::getProjectId, projectId)
            .eq(ApiRunningRecordEntity::getTaskStatus, ApiTaskStatus.success.getTaskStatus());
        long count1 = apiRunningRecordService.count(lambdaQueryWrapper);

        LambdaQueryWrapper<ApiRunningRecordEntity> lambdaQueryWrapper1 = new LambdaQueryWrapper<ApiRunningRecordEntity>().eq(
                ApiRunningRecordEntity::getProjectId, projectId)
            .eq(ApiRunningRecordEntity::getTaskStatus, ApiTaskStatus.failure.getTaskStatus());
        long count2 = apiRunningRecordService.count(lambdaQueryWrapper1);

        return new TaskOverviewDomain(toReleaseCount, count1, count2);
    }

    /**
     * 任务列表分页
     *
     * @param pageNum
     * @param pageSize
     * @param search
     * @return
     */
    @Override
    public PageResult<TaskItemDomain> getTaskPage(
        Integer pageNum, Integer pageSize, Integer status, String search) {
        //    select tb_api_http.* from tb_api_http left join tb_user on tb_api_http.creator_id =
        // tb_user.id where task_name like '%search%' or task_config_param like '%search%' or
        // tb_user.username like '%search%'
        Long projectId = ServletUtils.getProjectId();
        LambdaQueryWrapper<ApiHttpEntity> eq =
            new LambdaQueryWrapper<ApiHttpEntity>()
                .eq(ApiHttpEntity::getProjectId, projectId)
                .like(StringUtils.isNotBlank(search), ApiHttpEntity::getTaskName, search)
                .eq(ApiHttpEntity::getRowState, 1);
        if (Objects.nonNull(status) && !status.equals(-1)) {
            eq.eq(ApiHttpEntity::getTaskStatus, status);
        }
        eq.orderByDesc(ApiHttpEntity::getUpdateTime);
        Page<ApiHttpEntity> page = this.page(new Page<ApiHttpEntity>(pageNum, pageSize), eq);

        PageResult<TaskItemDomain> result =
            new PageResult<>(Lists.newArrayList(), pageNum, pageSize, 0);
        if (page == null || CollectionUtils.isEmpty(page.getRecords())) {
            return result;
        }
        result.setTotalElements(page.getTotal());
        result.setPageNum(pageNum);
        result.setPageSize(pageSize);
        result.setTotalPages(Math.toIntExact(page.getPages()));
        List<TaskItemDomain> resultDomainList = new ArrayList<>();
        List<String> userIdList =
            page.getRecords().stream().map(EngineBaseModel::getCreatorId).collect(Collectors.toList());
        Map<String, String> mapByUserIds = userService.getMapByUserIds(userIdList);

        for (ApiHttpEntity api : page.getRecords()) {
            TaskItemDomain targetPageInfo = new TaskItemDomain();
            targetPageInfo.setId(api.getId());
            targetPageInfo.setTaskName(api.getTaskName());
            targetPageInfo.setUrl(api.getApiHttpRequest().getUrl());
            targetPageInfo.setHttpRequestMethod(api.getApiHttpRequest().getHttpRequestMethod());
            TaskConfigParamEntity taskConfigParam = api.getTaskConfigParam();
            targetPageInfo.setStorageTable(taskConfigParam.getTargetTableName());
            targetPageInfo.setCycle(taskConfigParam.getCycle());
            targetPageInfo.setTaskStatus(api.getTaskStatus());
            targetPageInfo.setStartTime(api.getStartTime());
            targetPageInfo.setEndTime(api.getEndTime());

            // 转换成用户名
            String creatorId = api.getCreatorId() == null ? SUPER_ADMIN : api.getCreatorId();
            String userName = mapByUserIds.getOrDefault(creatorId, "");
            targetPageInfo.setCreateBy(userName);
            resultDomainList.add(targetPageInfo);
        }
        result.setResult(resultDomainList);
        return result;
    }

    /**
     * 任务运行后更改状态 和 启动时间
     *
     * @param id
     */
    @Override
    public void updateAfterStart(Long id) {
        ApiHttpEntity apiHttpEntity = new ApiHttpEntity();
        apiHttpEntity.setTaskStatus(ApiTaskStatus.running.getTaskStatus());
        apiHttpEntity.setStartTime(new Date());
        apiHttpEntity.setEndTime(null);
        this.update(
            apiHttpEntity, new LambdaUpdateWrapper<ApiHttpEntity>().eq(ApiHttpEntity::getId, id));
    }

    /**
     * 任务结束更新状态和结束时间
     */
    @Override
    public void updateAfterStop(Long logId, Integer status) {
        ApiRunningRecordDomain apiRunningRecord = apiRunningRecordService.getApiRunningRecord(logId);
        if (null == apiRunningRecord) {
            throw new BusinessException("任务运行记录不存在");
        }
        ApiHttpEntity apiHttpEntity = new ApiHttpEntity();
        apiHttpEntity.setTaskStatus(status);
        apiHttpEntity.setEndTime(new Date());
        this.update(
            apiHttpEntity,
            new LambdaUpdateWrapper<ApiHttpEntity>()
                .eq(ApiHttpEntity::getId, apiRunningRecord.getApiId()));
    }

    /**
     * 保存任务
     *
     * @param apiHttpSaveParam
     */
//    @BizDataIndexAnnotation(
//        type = BizDataIndexTypeEnum.API_HTTP_COLLECT,
//        dataIdFieldName = "id",
//        dataIdParamIndex = 0,
//        operateType = BizDataIndexAnnotation.OperateType.SAVE)
    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean save(ApiHttpSaveParam apiHttpSaveParam) {
        Long id = apiHttpSaveParam.getId();
        String taskName = apiHttpSaveParam.getTaskName();

        TaskConfigParam task1 = apiHttpSaveParam.getTask();
        if (Objects.nonNull(task1)) {
            Long targetDsId = task1.getTargetDsId();
            DataSourceDetailDomain dataSourceDetailById = dataSourceService.getDataSourceDetailById(
                targetDsId, false);
            Integer isSystem = dataSourceDetailById.getIsSystem();
            RdbmsConfig rdbmsConfig = dataSourceDetailById.getRdbmsConfig();
            if (Objects.nonNull(rdbmsConfig) && Objects.nonNull(isSystem) && isSystem.equals(1)) {
                String targetTableName = task1.getTargetTableName();
//                boolean secretLevel = tblPropService.checkSecretLevel(targetTableName,
//                    ServletUtils.getProjectId(), UserContext.getUserId());

//                if (!secretLevel) {
//                    userBehaviorLogService.addFailedLog(Business.接口采集, Operation.ADD_UPDATE,
//                        taskName,
//                        LogResult.FAILED);
//                    throw new BusinessException("密级不符合");
//                }
            }

        }

        Long projectId = ServletUtils.getProjectId();
        ApiHttpEntity one =
            this.getOne(
                new LambdaQueryWrapper<ApiHttpEntity>()
                    .eq(ApiHttpEntity::getProjectId, projectId)
                    .eq(ApiHttpEntity::getTaskName, taskName)
                    .eq(ApiHttpEntity::getRowState, 1));

        Long existId = null != one ? one.getId() : null;

        TestApiHttpParam testApiHttpParam = apiHttpSaveParam.getTestApiHttpParam();
        PreviewExpressParam previewParam = apiHttpSaveParam.getPreviewParam();
        TaskConfigParam task = apiHttpSaveParam.getTask();
        // tablename转换
        String originTableName = task.getTargetTableName();
        // datax json任务
//    DataX.JobContentReader jobContentReader = buildJobReader(testApiHttpParam, previewParam, task);
//    String jobJsonString = this.buildJobJson(jobContentReader, task);
        // 构建完任务后把schema去掉
        task.setTargetTableName(originTableName);
        ApiHttpEntity apiHttpEntity =
            buildApiHttpEntity(taskName, testApiHttpParam, previewParam, task, null);

        if (null == id) {
            if (null != existId) {
                throw new BusinessException("任务名称已存在");
            }
            apiHttpEntity.setTaskStatus(ApiTaskStatus.torelease.getTaskStatus());
            this.save(apiHttpEntity);
//            LogContext.put(LogKey.NAME.getKey(), apiHttpSaveParam.getTaskName());
            apiHttpSaveParam.setId(apiHttpEntity.getId());
        } else {
            if (null != existId && (!Objects.equals(existId, id))) {
                throw new BusinessException("任务名称已存在");
            }
            apiHttpEntity.setId(id);
            this.updateById(apiHttpEntity);
            if (null != apiHttpEntity.getXxlJobId()) {
                log.info("task更新,移除绑定的xxljob:{}", apiHttpEntity.getXxlJobId());
                xxlJobService.remove(apiHttpEntity.getXxlJobId());
            }
            // xxl job update不生效, 清空xxlJobId重新生成
            apiHttpEntity.setXxlJobId(null);
        }
        // 添加到xxlJob
        Integer xxlJobId;
        try {
            xxlJobId = this.generateJobId(apiHttpEntity, task);
        } catch (Exception e) {
            log.error("创建xxl任务失败{}", e.getMessage());
            throw new BusinessException("检查调度平台状态");
        }
        apiHttpEntity.setXxlJobId(xxlJobId);
        this.updateById(apiHttpEntity);
//        LogContext.put(LogKey.NAME.getKey(), apiHttpSaveParam.getTaskName());
//        LogContext.put(LogKey.ISUPDATE.getKey(), LogKey.ISUPDATE.getValue());

        return true;
    }

    @Override
    public String buildApiDataxJson(Long id, Map<String, String> param) {
        ApiHttpEntity byId = this.getById(id);
        ApiHttpRequest apiHttpRequest = byId.getApiHttpRequest();
        ApiResponseEntity apiResponse = byId.getApiResponse();
        TaskConfigParamEntity taskConfigParam = byId.getTaskConfigParam();
        TestApiHttpParam apiHttpParam = Convert.convert(TestApiHttpParam.class, apiHttpRequest);
        PreviewExpressParam previewParam = Convert.convert(PreviewExpressParam.class, apiResponse);
        TaskConfigParam task = Convert.convert(TaskConfigParam.class, taskConfigParam);
        if (MapUtils.isNotEmpty(param)) {
            ApiHttpEntity.ApiHttpRequestBodyParam apiHttpRequestBodyParam = apiHttpRequest.getApiHttpRequestBodyParam();
            List<FormParam> formParams = apiHttpRequestBodyParam.getFormParams();
            if (CollectionUtils.isEmpty(formParams)) {
                formParams = Lists.newArrayList();
            }
            Set<Entry<String, String>> entries = param.entrySet();
            Iterator<Entry<String, String>> iterator = entries.iterator();
            while (iterator.hasNext()) {
                Entry<String, String> next = iterator.next();
                String key = next.getKey();
                String value = next.getValue();
                FormParam formParam = new FormParam();
                formParam.setParamType(0);
                formParam.setParamName(key);
                formParam.setParamValue(value);
                formParam.setParamDataType("String");
                formParams.add(formParam);
            }
        }
        DataX.JobContentReader jobContentReader = buildJobReader(apiHttpParam, previewParam, task);
        String jobJsonString = this.buildJobJson(jobContentReader, task);
        return jobJsonString;
    }

    /**
     * 保存xxl任务生成任务id
     *
     * @param apiHttpEntity
     * @param task
     * @return
     */
    @Override
    public Integer generateJobId(ApiHttpEntity apiHttpEntity, TaskConfigParam task) {
        //    String glueSource = this.generateGlueSource(apiHttpEntity.getId());

        // 重用离线同步的调度配置
        OfflineSyncEntity.OfflineSyncSchedule schedule =
            ConvertUtil.copyProperties(task, OfflineSyncEntity.OfflineSyncSchedule.class);
        if (schedule == null) {
            return null;
        }

        int triggerStatus = 0;
        // 工作流未锁定且配置了调度,未发布
        if (Objects.nonNull(schedule.getStatus()) && schedule.getStatus().equals(0)) {
            triggerStatus = 1;
        }

        boolean isInsert = true;
        XxlJobInfo jobInfo = new XxlJobInfo();
        if (Objects.nonNull(apiHttpEntity.getXxlJobId())) {
            jobInfo.setId(Math.toIntExact(apiHttpEntity.getXxlJobId()));
            isInsert = false;
        }

        jobInfo.setJobGroup(XxlGroupEnum.DATA_ENGINE_SYSTEMTASK_GROUP.getId());

        jobInfo.setJobDesc(apiHttpEntity.getTaskName());

        final UserDomain userInfo = UserContext.getUserInfo();
        jobInfo.setAuthor(null == userInfo ? "admin" : userInfo.getUserName());
        jobInfo.setAlarmEmail(null == userInfo ? "" : userInfo.getEmail());

        jobInfo.setAddTime(new Date());

        jobInfo.setScheduleType(ScheduleTypeEnum.HUFU.name());

        // 设置未冻结
        schedule.setStatus(0);

        jobInfo.setScheduleConf(JSON.toJSONString(schedule));

        jobInfo.setMisfireStrategy(MisfireStrategyEnum.DO_NOTHING.name()); // 调度过期策略

        jobInfo.setExecutorRouteStrategy(ExecutorRouteStrategyEnum.ROUND.name());

        jobInfo.setExecutorHandler("OfflineApiHttpHandler");

        jobInfo.setExecutorBlockStrategy(ExecutorBlockStrategyEnum.DISCARD_LATER.name());

        jobInfo.setGlueType(GlueTypeEnum.BEAN.name());

        jobInfo.setTriggerStatus(triggerStatus); // 默认是停止状态

        jobInfo.setEffectFrom(schedule.getEffectiveDateStart());

        jobInfo.setEffectTo(schedule.getEffectiveDateEnd());

        if (isInsert) {
            ReturnT<String> add = xxlJobService.add(jobInfo);

            return Integer.parseInt(add.getContent());
        }
        ReturnT<String> add = xxlJobService.update(jobInfo);

        return Integer.parseInt(add.getContent());
    }

    @Override
    public Long getApiIdByXxlId(Long xxlId) {
        ApiHttpEntity one =
            this.getOne(
                new LambdaQueryWrapper<ApiHttpEntity>()
                    .eq(ApiHttpEntity::getXxlJobId, xxlId)
                    .eq(ApiHttpEntity::getRowState, 1));
        return Objects.isNull(one) ? null : one.getId();
    }

    /**
     * 获取任务详情
     *
     * @param taskId
     * @return
     */
    @Override
    public TaskDomain getTaskById(Long taskId) {
        ApiHttpEntity apiHttpEntity = this.getById(taskId);
        TaskDomain taskDomain = new TaskDomain();
        taskDomain.setId(apiHttpEntity.getId());
        taskDomain.setTaskName(apiHttpEntity.getTaskName());
        taskDomain.setDataxJobScript(apiHttpEntity.getDataxJobScript());
        taskDomain.setTaskStatus(apiHttpEntity.getTaskStatus());
        // 请求部分
        ApiHttpEntity.ApiHttpRequest sourceApiHttpRequest = apiHttpEntity.getApiHttpRequest();
        if (null != sourceApiHttpRequest) {
            TaskDomain.TestAPiHttpRequest testAPiHttpRequest = testApiHttpRequestMapper.map(sourceApiHttpRequest);
            testAPiHttpRequest.setApiHttpAuthParam(
                httpApiAuthParamMapper.map(sourceApiHttpRequest.getApiHttpAuthParam()));
            TaskDomain.TestAPiHttpRequest.ApiHttpRequestHeaderParam apiHttpRequestHeaderParam =
                new TaskDomain.TestAPiHttpRequest.ApiHttpRequestHeaderParam();

            TaskDomain.TestAPiHttpRequest.ApiHttpRequestBodyParam apiHttpRequestBodyParam =
                httpApiRequestBodyParamMapper.map(sourceApiHttpRequest.getApiHttpRequestBodyParam());
            if (CollectionUtils.isNotEmpty(
                sourceApiHttpRequest.getApiHttpRequestBodyParam().getFormParams())) {
                apiHttpRequestBodyParam.setFormParams(
                    ConvertUtil.copyProperties(
                        sourceApiHttpRequest.getApiHttpRequestBodyParam().getFormParams(),
                        TaskDomain.TestAPiHttpRequest.ApiHttpRequestBodyParam.FormParam.class));
            }
            testAPiHttpRequest.setApiHttpRequestBodyParam(apiHttpRequestBodyParam);

            taskDomain.setApiHttpRequest(testAPiHttpRequest);
        }

        // 返回值处理
        ApiResponseEntity sourceApiHttpResponse = apiHttpEntity.getApiResponse();
        if (null != sourceApiHttpResponse) {
            TaskDomain.PreviewExpressRequest previewExpressRequest =
                ConvertUtil.copyProperties(sourceApiHttpResponse, TaskDomain.PreviewExpressRequest.class);

            if (CollectionUtils.isNotEmpty(sourceApiHttpResponse.getExpressionParamList())) {
                previewExpressRequest.setExpressionParamList(
                    ConvertUtil.copyProperties(
                        sourceApiHttpResponse.getExpressionParamList(),
                        TaskDomain.PreviewExpressRequest.ExpressionParam.class));
            }
            taskDomain.setPreviewExpressRequest(previewExpressRequest);
        }
        TaskConfigParamEntity taskConfigParam = apiHttpEntity.getTaskConfigParam();
        // 任务配置相关
        TaskConfigDomain taskConfigRequest =
            ConvertUtil.copyProperties(taskConfigParam, TaskConfigDomain.class);
        taskConfigRequest.setTargetFields(
            ConvertUtil.copyProperties(
                taskConfigParam.getTargetFields(), TaskConfigDomain.TargetField.class));
        taskConfigRequest.setFieldMappings(
            ConvertUtil.copyProperties(
                taskConfigParam.getFieldMappings(), TaskConfigDomain.FieldMapping.class));
        taskConfigRequest.setPartitions(
            ConvertUtil.copyProperties(
                taskConfigParam.getPartitions(), TaskConfigDomain.PartitionField.class));
        taskDomain.setTaskConfigRequest(taskConfigRequest);
        return taskDomain;
    }

    /**
     * 参数转换到entity
     *
     * @param taskName
     * @param sourceApiHttpParam
     * @param sourcePreviewParam
     * @param sourceTask
     * @param jobJsonString
     * @return
     */
    @NotNull
    public ApiHttpEntity buildApiHttpEntity(
        String taskName,
        TestApiHttpParam sourceApiHttpParam,
        PreviewExpressParam sourcePreviewParam,
        TaskConfigParam sourceTask,
        String jobJsonString) {

        Long projectId = ServletUtils.getProjectId();
        ApiHttpEntity apiHttpEntity = new ApiHttpEntity();
        apiHttpEntity.setProjectId(projectId);
        apiHttpEntity.setTaskName(taskName);
        // http请求参数配置
        ApiHttpEntity.ApiHttpRequest apiHttpRequest =
            ConvertUtil.copyProperties(sourceApiHttpParam, ApiHttpEntity.ApiHttpRequest.class);

        ApiHttpEntity.ApiHttpAuthParam apiHttpAuthParam =
            ConvertUtil.copyProperties(
                sourceApiHttpParam.getApiHttpAuthParam(), ApiHttpEntity.ApiHttpAuthParam.class);
        apiHttpRequest.setApiHttpAuthParam(apiHttpAuthParam);

        // 复制body参数
        ApiHttpEntity.ApiHttpRequestBodyParam apiHttpRequestBodyParam =
            ConvertUtil.copyProperties(
                sourceApiHttpParam.getApiHttpRequestBodyParam(),
                ApiHttpEntity.ApiHttpRequestBodyParam.class);
        if (CollectionUtils.isNotEmpty(
            sourceApiHttpParam.getApiHttpRequestBodyParam().getFormParams())) {
            apiHttpRequestBodyParam.setFormParams(
                ConvertUtil.copyProperties(
                    sourceApiHttpParam.getApiHttpRequestBodyParam().getFormParams(),
                    ApiHttpEntity.ApiHttpRequestBodyParam.FormParam.class));
        }
        apiHttpRequest.setApiHttpRequestBodyParam(apiHttpRequestBodyParam);

        apiHttpEntity.setApiHttpRequest(apiHttpRequest);

        // 返回值解析
        ApiResponseEntity apiHttpResponse =
            ConvertUtil.copyProperties(sourcePreviewParam, ApiResponseEntity.class);
        if (CollectionUtils.isNotEmpty(sourcePreviewParam.getExpressionParamList())) {
            List<ApiResponseEntity.ExpressionParam> expressionParams =
                ConvertUtil.copyProperties(
                    sourcePreviewParam.getExpressionParamList(), ApiResponseEntity.ExpressionParam.class);
            apiHttpResponse.setExpressionParamList(expressionParams);
        }

        apiHttpEntity.setApiResponse(apiHttpResponse);

        // 任务数据目标,调度配置
        TaskConfigParamEntity taskConfigParam =
            ConvertUtil.copyProperties(sourceTask, TaskConfigParamEntity.class);
        taskConfigParam.setTargetFields(
            ConvertUtil.copyProperties(
                sourceTask.getTargetFields(), TaskConfigParamEntity.TargetField.class));
        taskConfigParam.setFieldMappings(
            ConvertUtil.copyProperties(
                sourceTask.getFieldMappings(), TaskConfigParamEntity.FieldMapping.class));
        taskConfigParam.setPartitions(
            ConvertUtil.copyProperties(
                sourceTask.getPartitions(), TaskConfigParamEntity.PartitionField.class));
        apiHttpEntity.setTaskConfigParam(taskConfigParam);

        apiHttpEntity.setDataxJobScript(jobJsonString);
        // 设置成未发布状态
        apiHttpEntity.setTaskStatus(ApiTaskStatus.torelease.getTaskStatus());
        apiHttpEntity.setRowState(1);
        apiHttpEntity.setCreateTime(new Date());
        apiHttpEntity.setCreatorId(UserContext.getUserId());
        apiHttpEntity.setUpdateTime(new Date());
        apiHttpEntity.setUpdaterId(UserContext.getUserId());
        return apiHttpEntity;
    }

    /**
     * 构建api reader job json
     *
     * @param testApiHttpParam
     * @param previewParam
     * @param task
     * @return
     */
    public DataX.JobContentReader buildJobReader(
        TestApiHttpParam testApiHttpParam, PreviewExpressParam previewParam, TaskConfigParam task) {
        DataX.JobContentReader reader = new DataX.JobContentReader();
        reader.setName(DataxReaderAndWriter.getReaderByName("Api"));

        ApiHttpJobContentReaderParameter readerParameter = new ApiHttpJobContentReaderParameter();

        readerParameter.setHttpUrl(testApiHttpParam.getUrl());
        readerParameter.setMethod(testApiHttpParam.getHttpRequestMethod());

        ApiHttpRequestBodyParam apiHttpRequestBodyParam = testApiHttpParam.getApiHttpRequestBodyParam();
        if (null != apiHttpRequestBodyParam) {
            String contentType = apiHttpRequestBodyParam.getContentType();
            // 设置请求方式,请求体
            readerParameter.setContentType(
                Objects.equals(testApiHttpParam.getHttpRequestMethod(), "POST") ? 1 : 0);
            readerParameter.setRaw(apiHttpRequestBodyParam.getBody());
            //    readerParameter.setTimeout(tableMapping.getRequestTimeout());
            readerParameter.setJsScript(previewParam.getJsScript());
        }

        List<ApiHttpJobContentReaderParameter.RequestParam> requestParams = Lists.newArrayList();

        // 设置auth
        ApiHttpAuthParam apiHttpAuthParam = testApiHttpParam.getApiHttpAuthParam();

        Headers headers = parseHeaderFromParam(Headers.of(), apiHttpAuthParam);
        for (int i = 0; i < headers.size(); i++) {
            ApiHttpJobContentReaderParameter.RequestParam item =
                new ApiHttpJobContentReaderParameter.RequestParam();
            item.setName(headers.name(i));
            item.setValue(headers.value(i));
            item.setPosition("head");
            requestParams.add(item);
        }

        readerParameter.setAdvancedFilePath(apiHttpAuthParam.getAdvancedFilePath());
        readerParameter.setAdvancedClassReference(apiHttpAuthParam.getAdvanceClassReference());
        readerParameter.setAdvancedMethodName(apiHttpAuthParam.getAdvanceMethodName());

        List<ApiHttpJobContentReaderParameter.ResponseParam> responseParams = Lists.newArrayList();

        List<String> fields = Lists.newArrayList();
        task.getFieldMappings()
            .forEach(
                i -> {
                    String sourceField = i.getSourceField();
                    fields.add(sourceField);
                });

        // 表单参数处理
        if (null != apiHttpRequestBodyParam
            && CollectionUtils.isNotEmpty(apiHttpRequestBodyParam.getFormParams())) {
            for (ApiHttpRequestBodyParam.FormParam param : apiHttpRequestBodyParam.getFormParams()) {
                Integer paramType = param.getParamType();
                if (paramType == 0) {
                    ApiHttpJobContentReaderParameter.RequestParam requestParam =
                        new ApiHttpJobContentReaderParameter.RequestParam();
                    requestParam.setName(param.getParamName());
                    requestParam.setPosition(param.getParamPosition());

                    Integer paramAttribute = param.getParamAttribute();
                    String type = null;
                    if (paramAttribute == 1) {
                        type = "pageNo";
                    } else if (paramAttribute == 2) {
                        type = "pageSize";
                    }
                    requestParam.setType(type);
                    // todo 系统参数值替换
                    requestParam.setValue(param.getParamValue());
                    requestParams.add(requestParam);
                }
            }
        }

        // 返回参数处理
        String jsScript = previewParam.getJsScript();
        if (StringUtils.isBlank(jsScript)) {
            List<ExpressionParam> expressionParamList = previewParam.getExpressionParamList();
            Map<String, ExpressionParam> fieldExpressionMap =
                expressionParamList.stream().collect(Collectors.toMap(ExpressionParam::getName, i -> i));

            // 用户直接建表, 维护表字段顺序匹配
            for (String sourceField : fields) {
                ExpressionParam expressionParam = fieldExpressionMap.get(sourceField);
                if (null == expressionParam) {
                    continue;
                }
                ApiHttpJobContentReaderParameter.ResponseParam responseParam =
                    new ApiHttpJobContentReaderParameter.ResponseParam();
                responseParam.setName(expressionParam.getName());
                responseParam.setExpression(expressionParam.getExpression());
                responseParam.setRelationship(expressionParam.getRelationship());
                responseParams.add(responseParam);
            }
        }

        readerParameter.setRequestParams(requestParams);
        readerParameter.setResponseParams(responseParams);

        reader.setParameter(readerParameter);
        return reader;
    }

    /**
     * 发送请求
     *
     * @param req
     * @return
     */
    @NotNull
    public TestApiHttpDomain execRequest(Request req) {
        OkHttpClient httpClient = new OkHttpClient.Builder()
            .readTimeout(500, TimeUnit.SECONDS)
            .sslSocketFactory(TLSUtil.createSSLSocketFactory(), new TLSUtil.TrustAllManager())
            .hostnameVerifier(new TLSUtil.TrustAllHostnameVerifier())
            .build();
        TestApiHttpDomain testResponse = new TestApiHttpDomain();
        boolean flagBreak = false;
        try (Response response = httpClient.newCall(req).execute()) {
            if (!response.isSuccessful()) {
                throw new BusinessException("接口返回值为空");
            }
            long totalBytesRead = 0;
            byte[] buffer = new byte[1024];
            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
            try (BufferedSource source = response.body().source()) {
                int bytesRead;
                while ((bytesRead = source.read(buffer)) != -1) {
                    totalBytesRead += bytesRead;
                    if (totalBytesRead > 1 * 1024 * 1024) {
                        flagBreak = true;
                        break;
                    }
                    outputStream.write(buffer, 0, bytesRead);
                }
            }
            // 将输出流中的数据转换为字符串
            String result = outputStream.toString();
            if (flagBreak) {
                List<String> jsonPaths = JsonPathUtil.getJsonPaths(result);
                testResponse.setExpressions(jsonPaths);
            }
            testResponse.setResponseBody(result);
        } catch (Exception e) {
            // 异常处理
        } finally {
            httpClient.connectionPool().evictAll();
        }
        return testResponse;
    }

    /**
     * 构造请求体
     *
     * @param url
     * @param httpRequestMethod
     * @param apiHttpRequestHeaderParam
     * @param contentType
     * @param formParams
     * @param rawJson
     * @return
     */
    @NotNull
    @Override
    public Request buildRequest(
        String url,
        String httpRequestMethod,
        ApiHttpAuthParam apiHttpAuthParam,
        ApiHttpRequestHeaderParam apiHttpRequestHeaderParam,
        String contentType,
        List<ApiHttpRequestBodyParam.FormParam> formParams,
        String rawJson) {

        Headers orginHeader = getHeaders(apiHttpRequestHeaderParam);

        Headers headers = parseHeaderFromParam(orginHeader, apiHttpAuthParam);

        List<ApiHttpRequestBodyParam.FormParam> requestParams = new ArrayList<>();

        final Map<String, String> advancedValueMap = this.exeAdvanced(apiHttpAuthParam);

        if (CollectionUtils.isNotEmpty(formParams)) {
            // 请求参数
            requestParams =
                formParams.stream()
                    .filter(x -> Objects.equals(x.getParamType(), 0))
                    .collect(Collectors.toList());

            try {

                for (ApiHttpRequestBodyParam.FormParam requestParam : requestParams) {
                    String value = requestParam.getParamValue();

                    if (null != value) {
                        final Matcher matcher = pattern.matcher(value);
                        while (matcher.find()) {
                            String key = matcher.group(1);
                            final String realVal = advancedValueMap.getOrDefault(key, null);
                            value = realVal;
                        }
                    }
                    requestParam.setParamValue(value);
                }
            } catch (Exception e) {
                throw new BusinessException(e.getMessage());
            }

            // 变量替换
            for (ApiHttpRequestBodyParam.FormParam requestParam : requestParams) {
                String paramValue = requestParam.getParamValue();
                String value = VarParserUtil.process(paramValue, null, null);
                requestParam.setParamValue(value);
            }
        }

        // 对raw做变量替换
        if (StringUtils.isNotBlank(rawJson)) {
            // 不再手动替换 pageNo, pageSize
            // rawJson = VarParserUtil.processForApiTaskTest(rawJson);
            StringSubstitutor sub = new StringSubstitutor(advancedValueMap);
            sub.setValueDelimiter("=");
            rawJson = sub.replace(rawJson);
        }

        String queryParam = this.getQueryParam(requestParams);
        Headers headerParam = this.getHeaderParam(requestParams);

        url += queryParam;

        Request.Builder builder = new Request.Builder().url(url);

        if (StringUtils.equalsIgnoreCase(httpRequestMethod, "GET")) {
            builder.get();
        } else {
            if (Objects.equals(ContentType.APPLICATION_JSON.getMimeType(), contentType)) {
                // 替换变量
                rawJson = VarParserUtil.processForApiTaskTest(rawJson);
                RequestBody jsonBodyParam = this.getJsonBodyParam(rawJson);
                builder.method(httpRequestMethod, jsonBodyParam);
            } else {
                RequestBody formBodyParam = this.getFormBodyParam(formParams);
                builder.method(httpRequestMethod, formBodyParam);
            }
        }

        if (null != headerParam) {
            Headers.Builder headBuilder = headers.newBuilder();
            Headers headerResult = headBuilder.addAll(headerParam).build();
            builder.headers(headerResult);
        }

        Request req = builder.build();
        return req;
    }

    @Override
    public List<ApiTaskDomain> getApiTaskDomainList(Long projectId) {

        List<ApiTaskDomain> rs = Lists.newArrayList();
        List<ApiHttpEntity> apiHttpEntities =
            this.list(
                new LambdaQueryWrapper<ApiHttpEntity>().eq(ApiHttpEntity::getProjectId, projectId));
        List<ApiTaskDomain> apiTaskDomains =
            ConvertUtil.copyProperties(apiHttpEntities, ApiTaskDomain.class);
        if (CollectionUtils.isNotEmpty(apiTaskDomains)) {
            rs.addAll(apiTaskDomains);
        }

        List<ApiWsEntity> apiWsEntities = apiWsService.getApiTaskByProjectId(projectId);
        List<ApiTaskDomain> apiWsDomain =
            ConvertUtil.copyProperties(apiWsEntities, ApiTaskDomain.class);
        if (CollectionUtils.isNotEmpty(apiWsDomain)) {
            rs.addAll(apiWsDomain);
        }
        return rs;
    }

    @Override
    public List<TaskDomain> getListByPage(Long projectId, Integer pageSize, Integer pageNum) {
        Page<ApiHttpEntity> page =
            this.page(
                new Page<>(pageNum, pageSize),
                new LambdaQueryWrapper<ApiHttpEntity>()
                    .eq(ApiHttpEntity::getProjectId, projectId)
                    .eq(ApiHttpEntity::getRowState, 1));
        return ConvertUtil.copyProperties(page.getRecords(), TaskDomain.class);
    }

    @Override
    public List<ApiHttpEntity> getApiHttpTaskEntityList(Long projectId) {
        List<ApiHttpEntity> result =
            this.list(
                new LambdaQueryWrapper<ApiHttpEntity>().eq(ApiHttpEntity::getProjectId, projectId)
                    .eq(ApiHttpEntity::getRowState, 1));
        return result;
    }

    /**
     * @param headers          api服务调用, 接口采集中不使用该参数请传Headers.of();
     * @param apiHttpAuthParam 认证参数
     * @return
     */
    private Headers parseHeaderFromParam(Headers headers, ApiHttpAuthParam apiHttpAuthParam) {
        if (null == headers) {
            headers = Headers.of();
        }
        if (null == apiHttpAuthParam) {
            return headers;
        }
        Headers.Builder headerBuilder = headers.newBuilder();
        // auth参数写入header
        switch (apiHttpAuthParam.getAuthType()) {
            case "noauth":
                break;
            case "basic":
                String basicAuth =
                    String.format(
                        "%s:%s", apiHttpAuthParam.getAuthUsername(), apiHttpAuthParam.getAuthPassword());
                headerBuilder
                    .add(
                        "Authorization",
                        "Basic "
                            + Base64.getEncoder()
                            .encodeToString(basicAuth.getBytes(StandardCharsets.UTF_8)))
                    .build();
                break;
            case "sign":
                headerBuilder
                    .add(apiHttpAuthParam.getAccessKey(), apiHttpAuthParam.getAccessSecret())
                    .build();
                break;
            case "token":
                headerBuilder.add("Authorization", "Bearer " + apiHttpAuthParam.getAccessToken()).build();
                break;
            case "jar":
                break;
            default:
                throw new BusinessException("不支持的认证方式");
        }

        headers = headerBuilder.build();
        return headers;
    }

    /**
     * 执行自定义jar
     *
     * @param apiHttpAuthParam
     * @return
     */
    private Map<String, String> exeAdvanced(ApiHttpAuthParam apiHttpAuthParam) {
        if (Objects.isNull(apiHttpAuthParam)) {
            return Collections.emptyMap();
        }
        if (!Objects.equals("jar",
            apiHttpAuthParam.getAuthType())) {
            return Collections.emptyMap();
        }

        String advancedFilePath = apiHttpAuthParam.getAdvancedFilePath();
        String advanceClassReference = apiHttpAuthParam.getAdvanceClassReference();
        String advanceMethodName = apiHttpAuthParam.getAdvanceMethodName();
        String advancedCacheKey = apiHttpAuthParam.getAdvancedCacheKey();
        Date expires = apiHttpAuthParam.getExpires();
        // 参数不能为空
        ArrayList<String> jarParameterList =
            Lists.newArrayList(
                advancedFilePath, advanceClassReference, advanceMethodName, advancedCacheKey);
        jarParameterList.stream()
            .findAny()
            .ifPresent(
                x -> {
                    if (StringUtils.isBlank(x)) {
                        throw new BusinessException("自定义jar参数不能为空");
                    }
                });

        try {
            // 获取jar包返回的变量,添加到缓存
            Map<String, String> entries = redisTemplate.opsForHash().entries(advancedCacheKey);

            if (entries.size() == 0) {
                Map<String, String> advancedMap =
                    this.execAdvanced(advancedFilePath, advanceClassReference, advanceMethodName);
                redisTemplate.opsForHash().putAll(advancedCacheKey, advancedMap);
                long exipreTime = expires.getTime() - System.currentTimeMillis();
                redisTemplate.expire(advancedCacheKey, exipreTime, TimeUnit.MILLISECONDS);
                return advancedMap;
            } else {
                return entries;
            }
            // 替换param中的变量
        } catch (Exception e) {
            throw new BusinessException("自定义jar执行失败");
        }
    }

    /**
     * 获取post请求参数
     *
     * @return
     */
    private RequestBody getFormBodyParam(List<ApiHttpRequestBodyParam.FormParam> requestParams) {
        FormBody.Builder builder = new FormBody.Builder();
        if (CollectionUtils.isEmpty(requestParams)) {
            return builder.build();
        }
        for (ApiHttpRequestBodyParam.FormParam param : requestParams) {
            String paramName = param.getParamName();

            String val = param.getParamValue();
            log.info("form param: {}={}", paramName, val);
            if (null != val) {
                builder.add(paramName, val);
            }
        }

        FormBody formBody = builder.build();

        return formBody;
    }

    /**
     * raw json
     *
     * @param json
     * @return
     */
    private RequestBody getJsonBodyParam(String json) {
        log.info("body json:{}", json);
        if (StrUtil.isBlank(json)) {
            throw new BusinessException("返回数据为空");
        }
        JSONObject jsonObject = JSONObject.parseObject(json);
        // 替换双引号
        if (Objects.nonNull(jsonObject)) {
            json = jsonObject.toJSONString();
        }

        log.info("body json after convert:{}", json);

        MediaType mediaType = MediaType.parse("application/json; charset=utf-8");

        return RequestBody.create(mediaType, json);
    }

    /**
     * header param
     *
     * @param param
     * @return
     */
    private Headers getHeaders(ApiHttpRequestHeaderParam param) {
        if (null == param || CollectionUtils.isEmpty(param.getParams())) {
            return Headers.of();
        }
        HashMap<String, String> headersMap = new HashMap<>();
        for (ApiHttpRequestHeaderParam.Header headerParam : param.getParams()) {
            headersMap.put(headerParam.getName(), headerParam.getValue());
        }
        return Headers.of(headersMap);
    }

    private Map<String, String> execAdvanced(
        String advancedFilePath, String advancedClassReference, String advancedMethodName)
        throws Exception {
        Map<String, String> resultMap = new HashMap<>();

        if (StringUtils.isBlank(advancedFilePath)
            || StringUtils.isBlank(advancedClassReference)
            || StringUtils.isBlank(advancedMethodName)) {
            log.info("advanced config will not be performed");
            return resultMap;
        }

        URL[] urls = new URL[]{};

        CustomerUrlClassLoader classLoader = new CustomerUrlClassLoader(urls);

        classLoader.addJar(new File(advancedFilePath).toURI().toURL());

        Class<?> clazz = classLoader.loadClass(advancedClassReference);

        Object instance = clazz.newInstance();

        Object returnValue = MethodUtils.invokeMethod(instance, advancedMethodName);

        // 清除相关引用供gc自动回收卸载Class
        instance = null;
        clazz = null;
        classLoader = null;

        if (null == returnValue || (!(returnValue instanceof HashMap))) {
            throw new BusinessException(
                "advanced config performed failed,the return is null or the return is not instanceof HashMap");
        }

        resultMap = (Map<String, String>) returnValue;

        return resultMap;
    }

    @Override
    public ApiLogDomain getOfflineSyncLog(Long logId, Integer offset) {
        return taskConfigServiceImpl.getOfflineSyncLog(logId, offset);
    }

    /**
     * 停止任务, 发stop请求到目标服务器上
     *
     * @param logId
     */
    @Override
    public void stop(Long logId) {
        ApiRunningRecordDomain apiRunningRecord = apiRunningRecordService.getApiRunningRecord(logId);
        String ipAddress = apiRunningRecord.getIpAddress();
        HashMap<String, String> paramMap = Maps.newHashMap();
        paramMap.put("logId", String.valueOf(logId));
        boolean canStop =
            distributedService.stopTaskPost(
                ipAddress, paramMap, OfflineDevTaskStopEnum.API_SYNC.getCode());
        if (!canStop) {
            dataXService.stopApiHttp(Long.valueOf(logId));
        }
    }

    @Override
    public Boolean ignoreApiRunningRecord(String ids) {
        if (StringUtils.isNotBlank(ids)) {
            List<Long> res = Arrays.stream(ids.split(",")).map(e -> Long.valueOf(e))
                .collect(Collectors.toList());
            List<ApiRunningRecordEntity> apiRunningRecordEntities = apiRunningRecordService.listByIds(
                res);
            apiRunningRecordEntities.forEach(e -> {
                e.setIgnored(1);
                e.setTaskStatus(
                    OfflineDevJobStatusEnum.IGNORED.getCode());
            });
            apiRunningRecordService.updateBatchById(apiRunningRecordEntities);
        }
        return true;
    }

    /**
     * 获取get请求参数
     *
     * @return
     */
    private String getQueryParam(List<ApiHttpRequestBodyParam.FormParam> requestParams) {
        if (CollectionUtils.isEmpty(requestParams)) {
            return "";
        }

        StringBuffer arg = new StringBuffer("?");

        for (ApiHttpRequestBodyParam.FormParam param : requestParams) {
            String paramPosition = param.getParamPosition();

            if (StringUtils.equalsIgnoreCase("query", paramPosition)) {
                String val = param.getParamValue();

                if (null != val) {
                    arg.append((param.getParamName()) + "=" + urlEncoderText(val) + "&");
                }
            }
        }

        return arg.deleteCharAt(arg.length() - 1).toString();
    }

    private Headers getHeaderParam(List<ApiHttpRequestBodyParam.FormParam> requestParams) {
        if (CollectionUtils.isEmpty(requestParams)) {
            return null;
        }

        Map<String, String> headers = new HashMap<>();

        for (ApiHttpRequestBodyParam.FormParam param : requestParams) {
            String paramPosition = param.getParamPosition();
            String paramName = param.getParamName();

            if (StringUtils.equalsIgnoreCase("head", paramPosition)) {
                String val = param.getParamValue();
                if (null != val) {
                    headers.put(paramName, val);
                }
            }
        }

        if (headers.isEmpty()) {
            return null;
        }

        return Headers.of(headers);
    }

    private String urlEncoderText(String text) {
        if (StringUtils.isBlank(text)) {
            return StringUtils.EMPTY;
        }
        try {
            text = URLEncoder.encode(text, "utf-8");
        } catch (UnsupportedEncodingException e) {
        }
        return text;
    }

    @Override
    public Integer getTaskStatus(Long jobId) {
        ApiRunningRecordDomain apiRunningRecord = apiRunningRecordService.getApiRunningRecord(jobId);
        return apiRunningRecord.getTaskStatus();
    }

    @Override
    public List<String> getNodeTableById(Long jobId) {
        TaskDomain taskById = this.getTaskById(jobId);
        TaskConfigDomain taskConfigRequest = taskById.getTaskConfigRequest();
        String targetTableName = taskConfigRequest.getTargetTableName();
        return Lists.newArrayList(targetTableName);
    }
}
